package spark.Task1;

import org.apache.spark.Partitioner;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

public class Task1_1 {
    public static void main( String[] args )
    {
        SparkConf conf = new SparkConf().setAppName("Task1").setMaster("local");
        try (JavaSparkContext task1_1 = new JavaSparkContext(conf)) {
            //读取csv
            JavaRDD<String> lines = task1_1.textFile("/home/jinhaitao/stock_small.csv");

            //将一行转化为键值对，键为年份+股票代码，值为交易数量
            JavaPairRDD<String, Long> KeyValuePair = lines.mapToPair(new PairFunction<String, String, Long>() {
                @Override
                public Tuple2<String, Long> call(String line) throws Exception {
                    String[] columns = line.split(",");
                    String key =  columns[2].substring(0,4) + "," + columns[1];
                    Long value = Long.parseLong(columns[7]);
                    Tuple2<String, Long> keyvalue = new Tuple2<>(key, value);
                    return keyvalue;
                }
            });

            //reduce操作将相同键的交易数量相加
            JavaPairRDD<String, Long> SumPair = KeyValuePair.reduceByKey(new Function2<Long, Long, Long>() {
                @Override
                public Long call(Long int1, Long int2) throws Exception {
                    Long result = int1 + int2;
                    return result;
                }
            });

            //排序
            JavaPairRDD<String, Long> SortPair = SumPair.mapToPair(row -> new Tuple2<Long,String>(row._2,row._1))      //键值换位
                                                                                        .sortByKey(false)                                                                                                      //对键倒序排
                                                                                        .mapToPair(row -> new Tuple2<String, Long>(row._2,row._1));                            //换位回来
    
            //SortPair.saveAsTextFile("/home/jinhaitao/result/Task1_4");
            JavaPairRDD<String, String>KeyValuePair2 = SortPair.mapToPair(new PairFunction<Tuple2<String, Long>, String, String>() {
                @Override
                public Tuple2<String, String> call(Tuple2<String, Long> t) throws Exception {
                    String new_key = t._1.substring(0,4);
                    String new_value = t._1.substring(5) + "," + Long.toString(t._2());
                    Tuple2<String, String> result = new Tuple2<>(new_key, new_value);
                    return result;
                }
            });
            

            //按年份分区以输入到不同文件
            JavaPairRDD<String, String> result = KeyValuePair2.partitionBy(new Partitioner(){
                @Override
                public int getPartition(Object key) {
                    return (Integer.parseInt((String) key) + 2) % 11;
                }
                @Override
                public int numPartitions() {
                    return 11;
                }
                });


            result.saveAsTextFile("/home/jinhaitao/result/Task1_1");

            task1_1.stop();
        }
        
    }
}
